package com.leonzhangxf;

import lombok.Value;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author leonzhangxf
 */
@Slf4j
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter, Runnable {

    private ConcurrentHashMap<String, Statistic> map = new ConcurrentHashMap<>();

    public TPMonitorFilter() {
        ScheduledExecutorService statisticExecutor =
            Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("statistic"));
        statisticExecutor.scheduleWithFixedDelay(this, 2000, 5000, TimeUnit.MILLISECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String key = String.format("%s_%s",
            invocation.getServiceName(), invocation.getMethodName());
        map.putIfAbsent(key, new Statistic());
        Statistic statistic = map.get(key);

        long start = System.currentTimeMillis();
        Result result = invoker.invoke(invocation);
        long end = System.currentTimeMillis();
        long cost = end - start;

        Node node = new Node(end, cost);
        statistic.addNode(node);

        return result;
    }

    @Override
    public void run() {
        // 统计TP90和TP99
        log.info("Print statistics start");
        for (Map.Entry<String, Statistic> entry : map.entrySet()) {
            log.info("Method: {}, TP90: {}, TP99: {}", entry.getKey(),
                entry.getValue().tp90(), entry.getValue().tp99());
        }
        log.info("Print statistics end");
    }

    public static class Statistic {

        List<Node> nodes = new ArrayList<>();

        public void addNode(Node node) {
            // 移除超过当前时间1分的数据
            long onMinutesAgo = System.currentTimeMillis() - 1000;
            List<Node> leftNodes = nodes.stream()
                .filter(n -> n.getTimestamp() - onMinutesAgo > 0)
                .collect(Collectors.toList());
            // 添加新的数据
            leftNodes.add(node);

            synchronized(this) {
                this.nodes = leftNodes;
            }
        }

        public long tp90() {
            return getTpPercent(0.9);
        }

        public long tp99() {
            return getTpPercent(0.99);
        }

        private long getTpPercent(double v) {
            List<Node> nodes;
            synchronized (this) {
                nodes = this.nodes;
            }
            int total = nodes.size();
            // 取百分比角标
            int index = Double.valueOf(Math.floor(total * v)).intValue();
            // 排序
            nodes.sort((n1, n2) -> Math.toIntExact(n1.getCost() - n2.getCost()));
            return nodes.get(index).getCost();
        }
    }

    @Value
    public static class Node {
        long timestamp;
        long cost;
    }
}
